package gupao.concurrency.simpledemo.risk.service;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class RiskThreadPoolService {

    private static final Log log = LogFactory.getLog(RiskThreadPoolService.class);

    private static RiskThreadPoolService INSTANCE;

    private static final String LEVEL_CALCULATOR_PARALLEL = "level.calculator.parallel";
    private static final String QAF_CALCULATOR_PARALLEL = "qaf.calculator.parallel";

    private static int defaultParallel;

    private final ExecutorService levelNodeThreadPool;

    private final ExecutorService nodeQafThreadPool;

    public synchronized static void init(){
        defaultParallel = Runtime.getRuntime().availableProcessors() + 1;
        log.info("init default parallel: " + defaultParallel);
        INSTANCE = new RiskThreadPoolService();
    }

    public static void showdownNow(){
        if(INSTANCE.levelNodeThreadPool != null){
            INSTANCE.levelNodeThreadPool.shutdownNow();
        }
        if(INSTANCE.nodeQafThreadPool != null){
            INSTANCE.nodeQafThreadPool.shutdownNow();
        }
    }

    public  RiskThreadPoolService() {
        int levelNodeCalcParallel = getConcurrencyProp(LEVEL_CALCULATOR_PARALLEL,defaultParallel);
        log.info("init Level Calculator Thread Pool with pool size: " + levelNodeCalcParallel);
        levelNodeThreadPool = new ThreadPoolExecutor(levelNodeCalcParallel, levelNodeCalcParallel,
                1L, TimeUnit.MINUTES,
                new ArrayBlockingQueue<Runnable>(1000),
                new SimpleThreadFactory("level-node-calc-"));

        int nodeQafCalcParallel = getConcurrencyProp(QAF_CALCULATOR_PARALLEL,defaultParallel);
        log.info("init Qaf Calculator Thread Pool with pool size: " + nodeQafCalcParallel);
        nodeQafThreadPool = new ThreadPoolExecutor(nodeQafCalcParallel, nodeQafCalcParallel,
                1L, TimeUnit.MINUTES,
                new ArrayBlockingQueue<Runnable>(1000),
                new SimpleThreadFactory("node-qaf-calc-"));
    }

    private int getConcurrencyProp(String propName, int defaultVal){
        String val = System.getProperty(propName);
        return val == null ? defaultVal : Integer.parseInt(val);
    }

    public ExecutorService getLevelNodeThreadPool() {
        return levelNodeThreadPool;
    }

    public ExecutorService getNodeQafThreadPool() {
        return nodeQafThreadPool;
    }

    public static RiskThreadPoolService getInstance(){
        return INSTANCE;
    }

    static class SimpleThreadFactory implements ThreadFactory{
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final String prefix;

        public SimpleThreadFactory(String prefix) {
            this.prefix = prefix;
        }

        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread( r,
                    prefix + threadNumber.getAndIncrement());
            t.setDaemon(true);
            if (t.getPriority() != Thread.NORM_PRIORITY)
                t.setPriority(Thread.NORM_PRIORITY);
            return t;
        }
    }
}
